package com.cgs.query.remote.netty

import com.cgs.query.exception.QueryException
import com.cgs.query.handler.Request
import com.cgs.query.handler.Response
import io.netty.bootstrap.Bootstrap
import io.netty.handler.timeout.IdleState
import io.netty.handler.timeout.IdleStateEvent
import io.netty.buffer.ByteBuf
import io.netty.channel.*
import io.netty.handler.codec.MessageToByteEncoder
import io.netty.handler.codec.ByteToMessageDecoder
import java.util.concurrent.TimeUnit
import io.netty.handler.timeout.IdleStateHandler
import io.netty.handler.logging.LoggingHandler
import io.netty.channel.socket.nio.NioSocketChannel
import java.net.InetSocketAddress
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.SocketChannel
import io.netty.handler.logging.LogLevel
import io.netty.util.concurrent.GenericFutureListener
import kotlinx.coroutines.experimental.delay
import kotlinx.coroutines.experimental.launch
import mu.KotlinLogging
import java.util.concurrent.atomic.AtomicInteger

private val logger = KotlinLogging.logger { }

class NettyClient(
        private val socketAddress: InetSocketAddress,
        private val serialization: Serialization) {

    private val group: EventLoopGroup = NioEventLoopGroup()

    private var channel: Channel? = null

    private val bootstrap = Bootstrap()

    private val listener = GenericFutureListener<ChannelFuture> {
        if (it.isSuccess) {
            logger.debug("连接成功$socketAddress")
        } else {
            throw QueryException("连接失败$socketAddress")
        }
    }

    @Throws(InterruptedException::class)
    fun start() {
        // 客户端
        bootstrap.group(group).channel(NioSocketChannel::class.java).handler(LoggingHandler(LogLevel.INFO))
                .option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true)

        bootstrap.handler(object : ChannelInitializer<SocketChannel>() {
            @Throws(Exception::class)
            override fun initChannel(ch: SocketChannel) {
                ch.pipeline().addLast(IdleStateHandler(0, 1, 0, TimeUnit.MINUTES),
                        NettyDecoder(Response::class.java), NettyEncoder(), NettyClientHandler())
            }
        })
        connect()
    }

    fun close() {
        group.shutdownGracefully(0, 5000, TimeUnit.SECONDS)
        if (channel != null && channel!!.isOpen) {
            channel!!.disconnect()
            channel!!.close()
        }
    }

    fun restart() {
        if (channel != null) {
            try {
                channel!!.close()
                connect()
            } catch (e: Exception) {
                logger.error{"重连失败：${e.message}"}
            }
        }
    }

    private fun connect() {
        channel = bootstrap.connect(socketAddress).sync().addListener(listener).channel()
    }

    fun send(request: Request) {
        this.channel!!.writeAndFlush(request)
    }

    internal inner class NettyDecoder(private val clzz: Class<*>) : ByteToMessageDecoder() {

        @Throws(Exception::class)
        override fun decode(ctx: ChannelHandlerContext, byteBuf: ByteBuf, out: MutableList<Any>) {
            if (byteBuf.readableBytes() < 4) {
                return
            }
            byteBuf.markReaderIndex()
            val dataLength = byteBuf.readInt()
            if (dataLength < 0) {
                ctx.close()
            }
            if (byteBuf.readableBytes() < dataLength) {
                byteBuf.resetReaderIndex()
                return  // fix 1024k buffer splice limix
            }
            val data = ByteArray(dataLength)
            byteBuf.readBytes(data)

            val obj = serialization.deserialize(data, clzz)
            out.add(obj)
        }
    }

    internal inner class NettyEncoder : MessageToByteEncoder<Request>() {

        @Throws(Exception::class)
        override fun encode(ctx: ChannelHandlerContext, msg: Request, out: ByteBuf) {
            val data = serialization.serialize<Any>(msg)
            out.writeInt(data.size)
            out.writeBytes(data)
        }
    }

    @ChannelHandler.Sharable
    internal inner class NettyClientHandler : ChannelInboundHandlerAdapter() {

        private val retryCount = AtomicInteger(1)
        private val ping = Request(0L, "ping", listOf(), String.javaClass)

        @Throws(Exception::class)
        override fun channelActive(ctx: ChannelHandlerContext) {
            super.channelActive(ctx)
            retryCount.set(1)
        }

        override fun channelInactive(ctx: ChannelHandlerContext?) {
            super.channelInactive(ctx)
            logger.error { "与服务端失去连接,进行重连" }

            launch {
                while (channel?.isActive == false && retryCount.getAndIncrement() < 6) {
                    delay(retryCount.get() * 3L, TimeUnit.SECONDS)
                    logger.debug { "第${retryCount.get() - 1}次重试" }
                    restart()
                }
                if (channel?.isActive == false) {
                    throw QueryException("重连失败").apply {
                        logger.debug { "重连失败,抛出异常" }
                    }
                }
            }
        }

        @Throws(Exception::class)
        override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
            if (msg is Response) {
                if (msg.requestId == 0L && logger.isDebugEnabled) {
                    print(msg.result)
                } else {
                    ResponseFuture.received(msg)
                }
            } else {
                logger.error { "错误返回类型:${msg.javaClass}" }
            }
        }

        @Throws(Exception::class)
        override fun userEventTriggered(ctx: ChannelHandlerContext, evt: Any) {
            if (evt is IdleStateEvent) {
                val state = evt.state()
                if (state == IdleState.WRITER_IDLE) {
                    logger.debug("ping")
                    ctx.writeAndFlush(ping)
                }
            } else {
                super.userEventTriggered(ctx, evt)
            }
        }
    }
}

